"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : security-audit
Case Name   : 嵌入打印函数的语句隐含了select操作应当被审计到
Create At   : 2023/06/08
@zou_jialiang0501162244
Description :
    1.设置审计参数，开启安全策略开关
    2.建表插入数据，创建打印函数
    3.配置脱敏策略
    4.创建索引并嵌入打印函数
    5.查询审计记录
Expect      :
    1.设置成功
    2.成功
    3.成功
    4.成功
    5.查询到创建FUNCTION、INDEX的信息(审计到对该表的select记录)
History     :
"""

import os
import unittest

from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant


class Audit(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.pri_node = Node('PrimaryDbUser')
        self.comm = Common()
        self.cons = Constant()
        self.resource_label = 'rl_audit_0011'
        self.masking_policy = 'mp_audit_0011'
        self.table = 't_audit_0011'
        self.func = 'f_audit_0011'
        self.para_value = self.comm.show_param('audit_system_object')

    def test_security(self):
        text = '-----step1:设置审计参数，开启安全策略开关;expect:成功-----'
        self.log.info(text)
        res = self.pri_sh.execut_db_sql(
            'alter system set audit_system_object=67121343;'
            'alter system set enable_security_policy = on;'
            'alter system set audit_dml_state_select = 1;')
        self.log.info(res)
        self.assertEqual(res.count(self.cons.alter_system_success_msg), 3,
                         f"执行失败{text}")

        text = '-----step2:建表插入数据，创建打印函数;expect:成功-----'
        self.log.info(text)
        sql = f'''select pg_delete_audit(
        sysdate - interval '1 minutes', sysdate);
        drop table if exists {self.table} cascade;
        create table {self.table}(id int, data varchar(100));
        insert into {self.table} values(1, 'foo');
        create function {self.func}(id integer,name varchar(100)) 
        returns integer language plpgsql IMMUTABLE as \$\$
        begin
        raise notice 'id :% , name : %',id,name;
        return 1;
        end;
        \$\$; '''
        msg = self.pri_sh.execut_db_sql(sql)
        self.log.info(msg)
        self.assertIn(self.cons.TABLE_CREATE_SUCCESS, msg, f"执行失败:{text}")
        self.assertIn(self.cons.INSERT_SUCCESS_MSG, msg, f"执行失败:{text}")
        self.assertIn(self.cons.CREATE_FUNCTION_SUCCESS_MSG, msg,
                      f"执行失败:{text}")

        text = '-----step3:配置脱敏策略;expect:成功-----'
        self.log.info(text)
        sql = f"drop resource label if exists {self.resource_label};" \
              f"drop masking policy if exists {self.masking_policy};" \
              f"create resource label {self.resource_label} " \
              f"add column({self.table}.data);" \
              f"create masking policy {self.masking_policy} randommasking " \
              f"on label({self.resource_label});"
        msg = self.pri_sh.execut_db_sql(sql)
        self.log.info(msg)
        self.assertTrue(self.cons.resource_label_create_success_msg in msg
                        and self.cons.masking_policy_create_success_msg in msg,
                        f"执行失败:{text}")

        text = '-----step4:创建索引并嵌入打印函数;expect:成功-----'
        self.log.info(text)
        msg = self.pri_sh.execut_db_sql(
            f"create index i_audit_0011 on {self.table}({self.func}(1,data));")
        self.log.info(msg)
        self.assertIn(self.cons.CREATE_INDEX_SUCCESS, msg, f"执行失败:{text}")

        text = '-----step5:查询审计记录;expect:审计到对该表的select记录-----'
        self.log.info(text)
        sql = f"select pg_sleep(3);" \
              f"select username,type,object_name,detail_info from " \
              f"pg_query_audit(sysdate - interval '20 seconds', sysdate) " \
              f"where type in('ddl_function','ddl_index','dml_action_select');"
        msg = self.pri_sh.execut_db_sql(sql)
        self.log.info(msg)
        self.assertIn(f'create function {self.func}', msg, f"执行失败:{text}")
        self.assertIn('create index i_audit_0011', msg, f"执行失败:{text}")

    def tearDown(self):
        text = '-----step6:清理环境;expect:成功-----'
        self.log.info(text)
        msg = self.pri_sh.execut_db_sql(
            f'drop table if exists {self.table} cascade;'
            f'drop masking policy if exists {self.masking_policy};'
            f'drop resource label if exists {self.resource_label};'
            f'drop function if exists {self.func}() cascade;')
        self.log.info(msg)
        res = self.pri_sh.execut_db_sql(
            f'alter system set audit_system_object={self.para_value};'
            f'alter system set enable_security_policy = off;'
            f'alter system set audit_dml_state_select = 0;')
        self.log.info(res)
        self.assertTrue(self.cons.drop_masking_policy_success in msg and
                        self.cons.drop_resource_label_success in msg,
                        f"执行失败{text}")
        self.assertIn(self.cons.TABLE_DROP_SUCCESS, msg, f"执行失败{text}")
        self.assertEqual(res.count(self.cons.alter_system_success_msg), 3,
                         f"执行失败{text}")
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
